package com.zzt.video.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.zzt.video.model.dto.RankListDto;
import com.zzt.video.model.dto.SeriesDto;
import com.zzt.video.model.dto.SeriesSearchParam;
import com.zzt.video.model.dto.SeriesShowDto;
import com.zzt.video.model.entity.Episode;
import com.zzt.video.model.entity.Series;
import com.zzt.video.model.mapper.EpisodeMapper;
import com.zzt.video.model.mapper.SeriesMapper;
import com.zzt.video.service.SeriesService;
import jakarta.servlet.http.HttpServletRequest;
import org.redisson.api.RAtomicLong;
import org.redisson.api.RBucket;
import org.redisson.api.RSet;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
public class SeriesServiceImpl extends ServiceImpl<SeriesMapper, Series> implements SeriesService {

    @Autowired
    private RedissonClient redissonClient; // 替换原来的 StringRedisTemplate

    @Autowired
    private EpisodeMapper episodeMapper;

    @Override
    public Page<Series> getSeriesByKeywordWithPagination(SeriesSearchParam param) {
        if(param.getCurrent() == null || param.getCurrent() <= 0){
            return null;
        }
        // 创建分页对象
        Page<Series> page = new Page<>(param.getCurrent(), 30L);

        // 调用 Mapper 方法进行模糊查询并按 update_time 降序
        return baseMapper.selectByKeyword(page, param);
    }

    @Override
    public SeriesShowDto getSeries(Long id) {
        Series series = baseMapper.selectById(id);
        if (series == null)
            return null;
        RBucket<List<SeriesDto>> seriesBucket = redissonClient.getBucket("series:latest:" + series.getType());
        List<SeriesDto> seriesList = seriesBucket.get();
        RBucket<List<RankListDto>> rankListBucket = redissonClient.getBucket("series:rank:daily:" + series.getType());
        List<RankListDto> rankListDtos = rankListBucket.get();
        SeriesShowDto dto = new SeriesShowDto(series);
        dto.setSeriesDtos(seriesList);
        dto.setRankListDtos(rankListDtos);
        String[] titles = {"日本视频","国产视频","自慰自拍","制服和cos","欧美"};
        dto.setCategoryTitle(titles[series.getType() - 1]);
        return dto;
    }

    @Override
    public void watchSeries(HttpServletRequest request, Long seriesId) {
        List<Episode> episodes = episodeMapper.selectList(
                new LambdaQueryWrapper<Episode>()
                        .eq(Episode::getSeriesId, seriesId)
        );
        if (episodes == null || episodes.isEmpty()) {
            return;
        }

        String ip = getClientIP(request);
        String today = LocalDate.now().format(DateTimeFormatter.BASIC_ISO_DATE); // 格式：yyyyMMdd

        // Key 定义
        String ipKey = "watch_log:" + episodes.get(0).getType() + ":ip:" + today;
        String seriesSetKey = "watch_log:" + episodes.get(0).getType() + ":series_set:" + today;
        String seriesKey = "watch_log:" + episodes.get(0).getType() + ":series:" + today + ":" + seriesId;

        // 记录 IP 当天访问了哪些视频
        RSet<String> ipSet = redissonClient.getSet(ipKey);
        ipSet.add(ip);

        // 记录该视频当天被访问过
        RSet<Long> seriesSet = redissonClient.getSet(seriesSetKey);
        seriesSet.add(seriesId);
        // 增加计数器
        RAtomicLong seriesCounter = redissonClient.getAtomicLong(seriesKey);
        seriesCounter.addAndGet(1);

        // 设置过期时间：40天
        ipSet.expire(40, TimeUnit.DAYS);
        seriesSet.expire(40, TimeUnit.DAYS);
        seriesCounter.expire(40, TimeUnit.DAYS);
    }
    private String getClientIP(HttpServletRequest request) {
        String ip = request.getHeader("X-Forwarded-For");
        if (ip == null || ip.isEmpty() || "unknown".equalsIgnoreCase(ip)) {
            ip = request.getRemoteAddr();
        }
        return ip;
    }

    @Override
    public List<Series> getLatestSeriesByType(int type, int limit) {
        QueryWrapper<Series> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("type", type)
                .orderByDesc("update_time")
                .last("LIMIT " + limit);

        return this.list(queryWrapper);
    }

    @Override
    public List<Series> getRandomSeriesByType(int type, int limit) {
        // 方案1: 使用数据库随机函数（适用于MySQL）
        QueryWrapper<Series> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("type", type)
                .last("ORDER BY RAND() LIMIT " + limit);

        return this.list(queryWrapper);
    }

    @Override
    public List<RankListDto> getDailyViewRankByType(int type, int limit) {
        String yesterday = LocalDate.now().minusDays(1).format(DateTimeFormatter.BASIC_ISO_DATE);
        String seriesSetKey = "watch_log:" + type + ":series_set:" + yesterday;

        RSet<Long> seriesSet = redissonClient.getSet(seriesSetKey);
        List<Long> seriesIds = new ArrayList<>(seriesSet.readAll());

        List<RankListDto> rankList = new ArrayList<>();

        // 只有当seriesIds不为空时才进行计数处理
        if (!seriesIds.isEmpty()) {
            String seriesKeyPrefix = "watch_log:" + type + ":series:" + yesterday;

            // 获取每个series的观看次数
            Map<Long, Long> viewCounts = new HashMap<>();
            for (Long seriesId : seriesIds) {
                RAtomicLong counter = redissonClient.getAtomicLong(seriesKeyPrefix + ":" + seriesId);
                viewCounts.put(seriesId, counter.get());
            }

            // 按照观看次数排序
            rankList = viewCounts.entrySet().stream()
                    .sorted(Map.Entry.<Long, Long>comparingByValue().reversed())
                    .map(entry -> {
                        RankListDto dto = new RankListDto();
                        dto.setId(entry.getKey());
                        return dto;
                    })
                    .collect(Collectors.toList());
        }

        // 如果数量不足或没有数据，用最新视频填充
        if (rankList.size() < limit) {
            QueryWrapper<Series> queryWrapper = new QueryWrapper<>();
            queryWrapper.eq("type", type)
                    .orderByDesc("update_time")
                    .last("LIMIT " + limit);

            List<Series> latestSeries = this.list(queryWrapper);
            Set<Long> existingIds = rankList.stream()
                    .map(RankListDto::getId)
                    .collect(Collectors.toSet());

            for (Series series : latestSeries) {
                if (rankList.size() >= limit) break;
                if (!existingIds.contains(series.getId())) {
                    RankListDto dto = new RankListDto();
                    dto.setId(series.getId());
                    rankList.add(dto);
                }
            }
        }
        List<Series> seriesList = this.list(new LambdaQueryWrapper<Series>()
                .in(Series::getId, rankList.stream().map(RankListDto::getId).toList())
        );
        Map<Long, Series> seriesMap = seriesList.stream().collect(Collectors.toMap(Series::getId, series -> series));
        rankList.forEach(dto -> dto.setSeriesName(seriesMap.get(dto.getId()).getSeriesName()));
        // 确保最终结果不超过limit
        return rankList.stream().limit(limit).collect(Collectors.toList());
    }


    @Override
    public List<RankListDto> getWeeklyViewRankByType(int type, int limit) {
        LocalDate today = LocalDate.now();
        List<String> dateKeys = new ArrayList<>();
        for (int i = 1; i <= 7; i++) {
            dateKeys.add(today.minusDays(i).format(DateTimeFormatter.BASIC_ISO_DATE));
        }

        Map<Long, Long> weeklyViewCounts = new HashMap<>();

        // 遍历过去7天的数据
        for (String date : dateKeys) {
            String seriesSetKey = "watch_log:" + type + ":series_set:" + date;

            RSet<Long> seriesSet = redissonClient.getSet(seriesSetKey);
            List<Long> seriesIds = new ArrayList<>(seriesSet.readAll());

            // 只有当seriesIds不为空时才进行计数处理
            if (!seriesIds.isEmpty()) {
                String seriesKeyPrefix = "watch_log:" + type + ":series:" + date;

                for (Long seriesId : seriesIds) {
                    RAtomicLong counter = redissonClient.getAtomicLong(seriesKeyPrefix + ":" + seriesId);
                    weeklyViewCounts.merge(seriesId, counter.get(), Long::sum);
                }
            }
        }

        List<RankListDto> rankList;

        if (!weeklyViewCounts.isEmpty()) {
            // 按照观看次数排序
            rankList = weeklyViewCounts.entrySet().stream()
                    .sorted(Map.Entry.<Long, Long>comparingByValue().reversed())
                    .map(entry -> {
                        RankListDto dto = new RankListDto();
                        dto.setId(entry.getKey());
                        return dto;
                    })
                    .collect(Collectors.toList());
        } else {
            // 如果一周内都没有观看记录，则初始化为空列表
            rankList = new ArrayList<>();
        }

        // 如果数量不足或没有数据，用最新视频填充
        if (rankList.size() < limit) {
            QueryWrapper<Series> queryWrapper = new QueryWrapper<>();
            queryWrapper.eq("type", type)
                    .orderByDesc("update_time")
                    .last("LIMIT " + limit);

            List<Series> latestSeries = this.list(queryWrapper);
            Set<Long> existingIds = rankList.stream()
                    .map(RankListDto::getId)
                    .collect(Collectors.toSet());

            for (Series series : latestSeries) {
                if (rankList.size() >= limit) break;
                if (!existingIds.contains(series.getId())) {
                    RankListDto dto = new RankListDto();
                    dto.setId(series.getId());
                    rankList.add(dto);
                }
            }
        }
        List<Series> seriesList = this.list(new LambdaQueryWrapper<Series>()
                .in(Series::getId, rankList.stream().map(RankListDto::getId).toList())
        );
        Map<Long, Series> seriesMap = seriesList.stream().collect(Collectors.toMap(Series::getId, series -> series));
        rankList.forEach(dto -> dto.setSeriesName(seriesMap.get(dto.getId()).getSeriesName()));
        // 确保最终结果不超过limit
        return rankList.stream().limit(limit).collect(Collectors.toList());
    }


}

